package org.budo.canal.message.handler;

import java.util.ArrayList;
import java.util.List;

import org.budo.support.lang.util.ListUtil;
import org.budo.support.lang.util.StringUtil;

import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;

import lombok.extern.slf4j.Slf4j;

/**
 * @author limingwei
 */
@Slf4j
public class DefaultCanalMessageHandler implements CanalMessageHandler {
    @Override
    public void handleMessage(Message message, ClientIdentity clientIdentity) {
        List<Entry> entries = message.getEntries();

        if (null == entries || entries.isEmpty()) {
            log.debug("#33 entries={}, message={}, clientIdentity={}", entries, message, clientIdentity);
            return;
        }

        List<Entry> entriesSkipNull = this.entriesSkipNull(message.getId(), entries);
        this.handleEntries(message.getId(), entriesSkipNull);
    }

    protected List<Entry> entriesSkipNull(Long messageId, List<Entry> entries) {
        List<Entry> list = new ArrayList<Entry>();
        for (Entry entry : entries) {
            Header header = entry.getHeader();

            String schemaName = header.getSchemaName();
            String tableName = header.getTableName();

            if (StringUtil.isEmpty(schemaName) || StringUtil.isEmpty(tableName)) {
                this.logEntryInfo(messageId, entry);

                continue;
            }

            list.add(entry);
        }

        return list;
    }

    /**
     * 覆盖这个方法
     */
    protected void handleEntries(Long messageId, List<Entry> entries) {
        for (Entry entry : entries) {
            this.logEntryInfo(messageId, entry);
        }
    }

    private void logEntryInfo(Long messageId, Entry entry) {
        Header header = entry.getHeader();

        String schemaName = header.getSchemaName();
        String tableName = header.getTableName();

        long serverId = header.getServerId(); // 可用于双向同步
        String logfileName = header.getLogfileName();
        long logfileOffset = header.getLogfileOffset();

        RowChange rowChange = RowChangeUtil.rowChangeParseFromByteString(entry.getStoreValue());
        EventType eventType = rowChange.getEventType();

        if (EventType.QUERY == eventType) {
            log.info("#42 QUERY == eventType, this=" + this + ", messageId=" + messageId + ", serverId=" + serverId //  
                    + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        // #42 QUERY, serverId=1, table=`warehouse_1`., logfileName=mysql-bin.000004,
        // logfileOffset=652, rowChange=rowChange.sql=CREATE DATABASE `warehouse_1`
        // CHARACTER SET utf8 COLLATE utf8_general_ci

        if (EventType.INSERT == eventType) {
            log.info("#48 INSERT, this=" + this + ", messageId=" + messageId + ", serverId=" + serverId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        if (EventType.DELETE == eventType) {
            log.info("#54 DELETE, this=" + this + ", messageId=" + messageId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        if (EventType.UPDATE == eventType && !ListUtil.isNullOrEmpty(rowChange.getRowDatasList())) {
            log.info("#62 UPDATE, this=" + this + ", messageId=" + messageId + ", serverId=" + serverId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        // rowChange.rowDatas is null
        if (EventType.UPDATE == eventType) {
            log.debug("#69 UPDATE, this=" + this + ", messageId=" + messageId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        log.warn("#65 eventType=" + eventType + ", this=" + this + ", messageId=" + messageId + ", serverId=" + serverId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset);
    }
}